package org.example.sink;

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.apache.flume.source.AbstractSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @ClassName MySink
 * @Author 小坏
 * @Date 2021/10/27、14:25
 * @Version 1.0
 */
public class MySink extends AbstractSink implements Configurable {

    private String prefix;
    private String subfix;

    private Logger logger = LoggerFactory.getLogger(MySink.class);

    @Override
    public void configure(Context context) {
        //前缀默认有一个
        prefix = context.getString("pre", "pre-");
        //后缀不指定就没了
        subfix = context.getString("sub");
    }

    @Override
    public Status process() throws EventDeliveryException {
        Channel channel = getChannel();
        Transaction transaction = channel.getTransaction();
        transaction.begin(); //开启事务


        //1、从 Channel中抓取数据打印控制台
        try {
            //抓取数据
            Event event;
            while (true) {
                event = channel.take();
                if (event != null) {
                    break;

                }
            }
            //2.2 处理数据
            logger.info(prefix + new String(event.getBody()) + subfix);

            //2.3 提交事务
            transaction.commit();
            return Status.READY;

        } catch (Exception e) {
            transaction.rollback();
            return Status.BACKOFF;
        } finally {
            transaction.close();
        }

    }
}
